package com.ljn;

import java.util.ArrayList;
import java.util.Arrays;
import org.apache.http.HttpHost;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestClientBuilder.HttpClientConfigCallback;
import org.elasticsearch.client.RestClientBuilder.RequestConfigCallback;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import com.google.common.collect.Lists;
import org.apache.http.client.config.RequestConfig.Builder;


@Configuration
public class ElasticSearchConfig implements FactoryBean<RestHighLevelClient>, InitializingBean, DisposableBean {

    @Value("${es.clusternodes}")
    private String clusterNodes;

    /*
     * private String clusterNodesIp; // ip地址 private Integer clusterNodesPort; // 端口号
     */
    private static int connectTimeOut = 1000; // 连接超时时间
    private static int socketTimeOut = 30000; // 连接超时时间
    private static int connectionRequestTimeOut = 500; // 获取连接的超时时间
    private static int maxConnectNum = 100; // 最大连接数
    private static int maxConnectPerRoute = 100; // 最大路由连接数

    private RestHighLevelClient restHighLevelClient;

    /**
     * 控制Bean的实例化过程
     *
     * @return
     * @throws Exception
     */
    @Override
    public RestHighLevelClient getObject() throws Exception {
        return restHighLevelClient;
    }

    /**
     * 获取接口返回的实例的class
     * 
     * @return
     */
    @Override
    public Class<?> getObjectType() {
        return RestHighLevelClient.class;
    }

    @Override
    public void destroy() throws Exception {
        try {
            if (restHighLevelClient != null) {
                restHighLevelClient.close();
            }
        } catch (final Exception e) {
           // log.error("Error closing ElasticSearch client: ", e);
        }
    }

    @Override
    public boolean isSingleton() {
        return false;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        // 取ip和port端口
        /*
         * clusterNodesIp = clusterNodes.split(":")[0]; clusterNodesPort = Integer.valueOf(clusterNodes.split(":")[1]);
         */

        restHighLevelClient = buildClient();
    }

    private RestHighLevelClient buildClient() {
        try {
            ArrayList<HttpHost> hosts = Lists.newArrayList();
            Arrays.stream(clusterNodes.split(",")).forEach(host -> {
                hosts.add(new HttpHost(host.split(":")[0], Integer.valueOf(host.split(":")[1]), "http"));
            });
            RestClientBuilder builder = RestClient.builder(hosts.toArray(new HttpHost[0]));

            /*
             * RestClientBuilder builder = RestClient.builder( new HttpHost(clusterNodesIp, clusterNodesPort, "http"));
             */

            // 异步httpclient连接延时配置
            builder.setRequestConfigCallback(new RequestConfigCallback() {
                @Override
                public Builder customizeRequestConfig(Builder requestConfigBuilder) {
                    requestConfigBuilder.setConnectTimeout(connectTimeOut);
                    requestConfigBuilder.setSocketTimeout(socketTimeOut);
                    requestConfigBuilder.setConnectionRequestTimeout(connectionRequestTimeOut);
                    return requestConfigBuilder;
                }
            });
            // 异步httpclient连接数配置
            builder.setHttpClientConfigCallback(new HttpClientConfigCallback() {
                @Override
                public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                    httpClientBuilder.setMaxConnTotal(maxConnectNum);
                    httpClientBuilder.setMaxConnPerRoute(maxConnectPerRoute);
                    return httpClientBuilder;
                }
            });

            restHighLevelClient = new RestHighLevelClient(builder);
        } catch (Exception e) {
//            log.error(e.getMessage());
        }
        return restHighLevelClient;
    }
}
